﻿using Sunxsh.Es.Common.Util;
using Newtonsoft.Json;
using RabbitMQ.Client;

namespace Sunxsh.Es.Common.Rabbitmq;

/// <summary>
/// 扇形交换机连接
/// </summary>
/// <typeparam name="T"></typeparam>
public class MQFanoutClient<T>
{

    MQConn _conn;
    string _exchangeName;
    string _queueName;

    IModel _channel;

    Task _initTask;
    Task _waitTask;

    internal MQFanoutClient(MQConn conn, string name)
    {
        _conn = conn;
        _exchangeName = name;

        _initTask = Task.Run(connect);                          //连接服务器的任务， 会一直重试
        _waitTask = Task.WhenAny(_initTask, Task.Delay(15_000)); //发送时候等待连接，最多等待15秒
    }


    public MQFanoutClient(string name)
        : this(MQConn.Huasky, name)
    {
    }


    async Task connect()
    {
        var channel = await _conn.CreateChannel();
        lock (this)
        {
            channel.ExchangeDeclare(_exchangeName, ExchangeType.Fanout, true, false, null);
            var queue = channel.QueueDeclare();
            channel.QueueBind(queue.QueueName, _exchangeName, "", null);
            _channel = channel;
            _queueName = queue.QueueName;
        }
    }

    /// <summary>
    /// 给队列发送消息
    /// </summary>
    public void Publish(T data)
    {
        if (!_waitTask.IsCompleted)
        {
            _waitTask.Wait();
        }

        if (_channel == null)
        {
            LoggerHelper.Error("MQ 未连接 [{0}]", _exchangeName);
            throw new InvalidOperationException("MQ 未连接");
        }

        lock (this) //线程安全
        {
            _channel.Publish(_exchangeName, "", data);
        }
    }

    /// <summary>
    /// 从队列获取消息， 多个进程注册同一个队列为多消费者模型
    /// 同一个进程多次注册同一个队列的回调会报错
    /// </summary>
    public async void Subscribe(ushort fetchSize, Func<T, Task> onData)
    {
        await _initTask;
        lock (this)
        {
            _channel.Subscribe(_queueName, fetchSize, para =>
            {
                var data = JsonConvert.DeserializeObject<T>(para.Content);
                return onData(data);
            });
        }
    }

}
